<?php
require_once SzSystem::$FRAME_ROOT . '/vendor/autoload.php';

class SzKafKaLogger extends SzAbstractLogger
{
    /**
     * @var \Kafka\Producer
     */
    private $producer;

    /**
     * @var string
     */
    private $topic;
    /**
     * @var array
     */
    private $buffer;
    /**
     * @see SzAbstractLogger::__construct
     * @param int $level
     * @param string $topic
     * @param string $host kafka 地址, eg: ip:port
     * @param string $version kafka 版本号
     */
    public function __construct($level = null, $topic = '', $host = '127.0.0.1:9092', $version = '0.10.1.0')
    {
        parent::__construct($level);

        $config = \Kafka\ProducerConfig::getInstance();
        $config->setMetadataRefreshIntervalMs(10000);
        $config->setMetadataBrokerList($host);
        $config->setBrokerVersion($version);
        $config->setRequiredAck(1);
        $config->setIsAsyn(false);
        $config->setProduceInterval(500);

        $this->producer = new \Kafka\Producer();
        $this->topic = $topic;
        $this->buff = [];
    }

    /**
     * @see SzAbstractLogger::doLog
     */
    protected function doLog($level, $message, $params = null)
    {
        if (is_object($message) || is_array($message)) {
            $message = var_export($message, true);
        }
        if ($this->producer) {
            $msg = [
                'topic' => $this->topic,
                'value' => $this->formatMessage($level, $message, $params)
            ];
            // 数据埋点统一发送
            if (isset($params['channel']) &&
                $params['channel'] == SzAbstractLogger::LOG_CHANNEL_BUSINESS) {
                $this->buffer[] = $msg;
            } else {
                $this->producer->send([$msg]);
            }
        }
        return true;
    }

    /**
     * @see SzAbstractLogger::closeLog
     */
    public function closeLog()
    {

    }

    /**
     * 发送日志
     * @return void
     */
    public function sendLog()
    {
        if (!$this->buffer) {
            return; // no log to send
        }
        if ($this->producer) {
            $this->producer->send($this->buffer);
            $this->buffer = array();
        }
    }
}